package cn.itcast.watermaker;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;
import java.util.Random;
import java.util.UUID;

/**
 * @author KTL
 * @version V1.0
 * @Package cn.itcast.watermaker
 * @date 2021/2/27 0027 16:35
 * @Copyright © 2015-04-29  One for each, and two for each
 * Desc 演示基于事件时间的窗口计算+Watermaker解决一定程度上的数据乱序/延迟到达问题
 */
public class WatermakerDemo01 {
    public static void main(String[] args) throws Exception {
        //TODO 0 ENV
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //TODO 1 SOURCE
        final DataStreamSource<Order> orderDS = env.addSource(new SourceFunction<Order>() {
            private boolean flag = true;

            @Override
            public void run(SourceContext<Order> ctx) throws Exception {
                Random random = new Random();
                while (flag) {
                    final String orderId = UUID.randomUUID().toString();
                    final int userId = random.nextInt(2);
                    final int money = random.nextInt(101);
                    //随机模拟延迟
                    long eventTime = System.currentTimeMillis() - random.nextInt(5) * 1000;
                    ctx.collect(new Order(orderId, userId, money, eventTime));
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {
                flag = false;
            }
        });
        //TODO 2 tranfromat
        //每隔5s计算最近5s的数据求每个用户的订单总金额，要求：基于事件时间进行窗口计算+watermaker
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);  // 新版本中默认就是EventTime
        final SingleOutputStreamOperator<Order> orderSingleOutputStreamOperator = orderDS.assignTimestampsAndWatermarks(WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(3)) //指定maxOutOfOrderness最大无序度/最大允许延迟时间/乱序时间
                .withTimestampAssigner((order, timestamp) -> order.getEventTime())
        );
        final SingleOutputStreamOperator<Order> result = orderSingleOutputStreamOperator.keyBy(Order::getUserId).window(TumblingEventTimeWindows.of(Time.seconds(5))).sum("money");
        //TODO 3 SINK
        result.print();
        //TODO 4 excuter
        env.execute();

    }
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public  static class  Order{
        private String orderId;
        private Integer userId;
        private Integer money;
        private Long eventTime;
    }
}
